/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.others.dstream;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Scanner;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

import lac.cnet.sddl.objects.ApplicationObject;
import lac.cnet.sddl.objects.Message;
import lac.cnet.sddl.udi.core.UniversalDDSLayerFactory;
import lac.cnet.sddl.udi.core.listener.UDIDataReaderListener;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.log4j.ConsoleAppender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;

import br.pucrio.inf.lac.konoha.hyuga.common.Geo;
import br.pucrio.inf.lac.konoha.hyuga.common.Grid;
import br.pucrio.inf.lac.konoha.hyuga.common.Pair;
import br.pucrio.inf.lac.konoha.hyuga.core.DDSLink;
import br.pucrio.inf.lac.konoha.hyuga.datastructure.SegmentTree;
import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.others.dstream.GridCharacteristicVector.STATUS;
import br.pucrio.inf.lac.konoha.hyuga.others.dstream.GridCharacteristicVector.TYPE;

import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * A D-Stream grid clustering implementation.
 * 
 *  TODO: Write JavaDoc.
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class DStream implements UDIDataReaderListener<ApplicationObject> {

    /** Algorithm parameters. */
    private Properties properties;
    
    /** D-Stream Density Thresholds. */
    private int lowPts;
    private int timeWindow;
    private int batchWindow;
    
    // DBSCAN inputs
    private int minPts;
    private long eps;
    
    // Domain data
    private double minLng;
    private double maxLng;
    private double minLat;
    private double maxLat;

    /** Grid Feature. */
    private GridCharacteristicVector[][] grid;
    
    private Map<Pair<Integer,Integer>, GridCharacteristicVector> cellList;
    /** Grid List. */
    
    /** Grid Cluster. */
    private List<GridClusterDStream> gridClusters;
    
    /** Input Stream. */
    private DDSLink dataStream;
    
    /** Input Queue. */
    private BlockingDeque<LocationUpdateEvent> eventStream;
    
    /** Output. */
    private Map<Long, List<DStreamOutput>> output;

    /** SegmentTree for latitude and longitude mapping. */
    private SegmentTree latitudeSegments;
    private SegmentTree longitudeSegments;
    
    /** JSON Mapper. */
    private ObjectMapper mapper;
    
    /** LOGGER. */
	private static Logger LOGGER = Logger.getLogger("br.pucrio.inf.lac.konoha.hyuga");
	
	private long initialTimestamp;
	private boolean hasSetInitialTimestamp;
	
	/** Default constructor load the algorithm parameters from dstream.properties. */
	public DStream(String fileName) {
		try {
		    // Configuration
	        properties = new Properties();
	        FileInputStream fi = new FileInputStream(fileName);
	        properties.load(fi);
	        fi.close();

		    // Parameters	
            minLat = Double.parseDouble(properties.getProperty("minlat"));
            maxLat = Double.parseDouble(properties.getProperty("maxlat"));
            minLng = Double.parseDouble(properties.getProperty("minlng"));
            maxLng = Double.parseDouble(properties.getProperty("maxlng"));
            eps    = Long.parseLong(properties.getProperty("eps"));
            minPts = Integer.parseInt(properties.getProperty("minPts"));
            lowPts = Integer.parseInt(properties.getProperty("lowPts"));
            timeWindow  = Integer.parseInt(properties.getProperty("win"));      // Consider the nodes in the win
            batchWindow = Integer.parseInt(properties.getProperty("batchwin")); // We are going to batch every batchwin second
           
            // Initializing D-Stream Variables
            preparePartitionMapper(minLat, maxLat, minLng, maxLng, eps);
            int lngsize = longitudeSegments.getSegments().length;
            int latsize = latitudeSegments.getSegments().length;
            
            int batchSize = timeWindow/batchWindow;
            gridClusters = new ArrayList<GridClusterDStream>();
            grid = new GridCharacteristicVector[lngsize][latsize];
            for (int i = 0; i < lngsize; i++) {
                for (int j = 0; j < latsize; j++) {
                    grid[i][j] = new GridCharacteristicVector(i, j, batchSize);
                }
            }
            
            eventStream = new LinkedBlockingDeque<LocationUpdateEvent>();
            output = new TreeMap<Long, List<DStreamOutput>>();
            hasSetInitialTimestamp = false;
            
            // Utilities
            mapper = new ObjectMapper();
            System.out.println("BATCH Window: " + batchWindow);
            System.out.println("TIME Window: " + timeWindow);
		} catch (IOException e) {
			System.err.println("Problem when loading dstream.properties");
			System.exit(-1);
		}
	}
	
    private void preparePartitionMapper(double minLat, double maxLat, double minLng, double maxLng, long eps) {
        ArrayList<Double> latitudeValues  = new ArrayList<Double>();
        ArrayList<Double> longitudeValues = new ArrayList<Double>();

        // Segments intervals, [minLat, minLat + ε, ..., maxLat] and [minLng, minLng + ε, ..., maxLng]
        double currentLat = minLat;
        double currentLng = minLng;
        do {
            if (currentLat < maxLat) latitudeValues.add(currentLat);
            if (currentLng < maxLng) longitudeValues.add(currentLng);

            currentLng = Geo.getOffsetLongitude(currentLat, currentLng, eps);
            currentLat = Geo.getOffsetLatitude(currentLat, eps);
        } while ((currentLat < maxLat) || (currentLng < maxLng));
        
        /* Add lower and higher bounds (-90 and 90 degrees for lat) and (-180 and 180 for longitude),
         * for values outside our monitoring space, in the first and last position of the intervals. */
        latitudeValues.add(0, (double) -90);
        latitudeValues.add((double) 90);

        longitudeValues.add(0, (double) -180);
        longitudeValues.add((double) 180);

        latitudeSegments  = new SegmentTree(ArrayUtils.toPrimitive(latitudeValues.toArray(new Double[latitudeValues.size()])));
        longitudeSegments = new SegmentTree(ArrayUtils.toPrimitive(longitudeValues.toArray(new Double[longitudeValues.size()])));
    }

    @Override
    public void onNewData(ApplicationObject data) {
        if (!hasSetInitialTimestamp) {
            initialTimestamp = System.currentTimeMillis();
            hasSetInitialTimestamp = true;
        }
        try {
            String locationUpdateJSON = new String(((Message) data).getContent());
            LocationUpdateEvent locationUpdateEvent = mapper.readValue(locationUpdateJSON, LocationUpdateEvent.class);

            int x = longitudeSegments.findIndex(locationUpdateEvent.getLongitude());
            int y = latitudeSegments.findIndex(locationUpdateEvent.getLatitude());
            locationUpdateEvent.setX(x);
            locationUpdateEvent.setY(y);
            
            eventStream.add(locationUpdateEvent);
        } catch (IOException ex) {
            System.err.println("Message not supported.");
        }        
    }
    
	/** Initiate D-Stream */
	public void cluster() {
		// Data Stream
		dataStream  = DDSLink.getInstance(UniversalDDSLayerFactory.convertStrToSupportedDDSVendor(properties.getProperty("dds")));
		dataStream.createSpatialTopic("LocationUpdateEvent");
		dataStream.addListener("LocationUpdateEvent", this);

		// Main 
        long currentTime   = System.currentTimeMillis();
        long timeThreshold = System.currentTimeMillis();
        boolean initialCluster = false;

        cellList = new HashMap<Pair<Integer,Integer>, GridCharacteristicVector>();
        Pair<Integer, Integer> cellIndex = null;
		while (true) {
		    timeThreshold = currentTime + (batchWindow * 1000);
		    
		    do {
		        try {
                    LocationUpdateEvent lu = eventStream.takeFirst();
                    GridCharacteristicVector cell = grid[lu.getX()][lu.getY()];

                    cellIndex = new Pair<Integer, Integer>(lu.getX(), lu.getY());
                    
                    if (!cellList.containsKey(cellIndex)) {
                        cellList.put(cellIndex, cell);
                    }
                    
                    cell = cellList.get(cellIndex);
                    cell.updateCharacteristicVector(lu);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
		        currentTime = System.currentTimeMillis();		        
		    } while (currentTime < timeThreshold);

		    if (!initialCluster) {
		        initialClustering();
		        initialCluster = true;
		    } else {
		        adjustClustering();
		    }
		}
	}

	private void initialClustering() {
	    for (GridCharacteristicVector cell : cellList.values()) {
            cell.updateAndResetBatch();
            
            double density = cell.getDensity();

            if (density >= minPts) {
                cell.setType(TYPE.DENSE);
                gridClusters.add(new GridClusterDStream(cell, getNeighbors(cell)));
            } else if (density >= lowPts) {
                cell.setType(TYPE.TRANSITIONAL);
            } else {
                cell.setType(TYPE.SPARSE);
            }
            
            cell.saveAndResetStatus();
        }
	    System.out.println("-----");
        boolean modified = false;
        do {
            modified = false;
            ArrayList<GridClusterDStream> neighborList = new ArrayList<GridClusterDStream>();
            
            for (GridClusterDStream cluster : gridClusters) {
                for (GridClusterDStream otherCluster : gridClusters) {
                    if (Grid.isNeighborDStream(cluster, otherCluster)) {
                        cluster.merge(otherCluster);
                        neighborList.add(otherCluster);
                        modified = true;
                    }
                }
                
                if (modified) break;
            }
            
            if (neighborList.size() != 0) {
                gridClusters.removeAll(neighborList);
            }
        } while (modified);
        
        long diffTimestamp = 60900000 + (System.currentTimeMillis() - initialTimestamp);
        diffTimestamp = diffTimestamp / 1000;
        diffTimestamp = (diffTimestamp + 1) * 1000;
        
        ArrayList<DStreamOutput> outList = new ArrayList<DStream.DStreamOutput>();
        for (GridClusterDStream cluster : gridClusters) {
            System.out.println(diffTimestamp + " ~ " + cluster);
            outList.add(new DStreamOutput(cluster.size(), cluster.getNodes()));
        }
        output.put(diffTimestamp, outList);
        System.out.println("-----");
	}
	
	
	private void adjustClustering() {
        // detect and remove sporadic grids from grid_list
        Iterator<Pair<Integer,Integer>> it = cellList.keySet().iterator();
        while (it.hasNext()) {
            GridCharacteristicVector cell = cellList.get(it.next());
            cell.updateAndResetBatch();
            double density = cell.getDensity();
            
            if (density > minPts) {
                cell.setType(TYPE.DENSE);
                cell.setStatus(STATUS.NORMAL);
            } else if (density > lowPts){
                cell.setType(TYPE.TRANSITIONAL);
                cell.setStatus(STATUS.NORMAL);
            } else {
                cell.setType(TYPE.SPARSE);
                if (cell.getPrevStatus() == STATUS.SPORADIC) {
                    if (!cell.hasReceivedData()) {
                        cell.clear();
                        it.remove();
                        continue;
                    }
                } else {
                    cell.setStatus(STATUS.SPORADIC);
                }
            }
        }
        
        for (GridCharacteristicVector cell : cellList.values()) {
            if (cell.hasChangedClass()) {
                if (cell.getPrevType() == TYPE.DENSE && cell.getType() != TYPE.DENSE) {
                    GridClusterDStream cluster = getClusterContainCell(cell);
                    
                    cluster.remove(cell);
                    cluster.recalculateNeighbors();
                    
                    List<Set<Pair<Integer, Integer>>> remainderClusters = cluster.getRemainderClusters();
                    if (remainderClusters.size() == 0) {
                        gridClusters.remove(cluster);
                    } else if (remainderClusters.size() > 1) {
                        gridClusters.remove(cluster);
                        
                        for (Set<Pair<Integer,Integer>> aCluster : remainderClusters) {
                            GridClusterDStream residualCluster = new GridClusterDStream();
                            
                            for (Pair<Integer,Integer> clusterIndex : aCluster) {
                                int x = clusterIndex.getFirst();
                                int y = clusterIndex.getSecond();
                                residualCluster.merge(new GridClusterDStream(grid[x][y], getNeighbors(grid[x][y])));
                            }
                            gridClusters.add(residualCluster);
                        }
                    }
                } else if (cell.getPrevType() != TYPE.DENSE && cell.getType() == TYPE.DENSE) {
                    if (!hasNeighbor(cell)) {
                        gridClusters.add(new GridClusterDStream(cell, getNeighbors(cell)));
                    } else {
                        GridClusterDStream neighborCluster = getNeighborCluster(cell);
                        neighborCluster.merge(new GridClusterDStream(cell,getNeighbors(cell)));
                    }
                    
                }
            }
           cell.saveAndResetStatus();
        }
        
        long diffTimestamp = 60900000 + (System.currentTimeMillis() - initialTimestamp);
        diffTimestamp = diffTimestamp / 1000;
        diffTimestamp = (diffTimestamp + 1) * 1000;
        
        ArrayList<DStreamOutput> outList = new ArrayList<DStream.DStreamOutput>();
        for (GridClusterDStream cluster : gridClusters) {
            System.out.println(diffTimestamp + " ~ " + cluster);
            outList.add(new DStreamOutput(cluster.size(), cluster.getNodes()));
        }
        
        output.put(diffTimestamp, outList);
        System.out.println("-----");
	}

    private GridClusterDStream getNeighborCluster(GridCharacteristicVector cell) {
        GridClusterDStream neighbor = null;

        for (GridClusterDStream cluster : gridClusters) {
            if (cluster.isNeighbor(cell)) {
                neighbor = cluster;
                break;
            }
        }

        return neighbor;
    }
	
	private boolean hasNeighbor(GridCharacteristicVector cell) {
	    boolean neighbor = false;
	    
	    for (GridClusterDStream cluster : gridClusters) {
            if (cluster.isNeighbor(cell)) {
                neighbor = true;
                break;
            }
        }
	    
	    return neighbor; 
	}
	
	private GridClusterDStream getClusterContainCell(GridCharacteristicVector cell) {
	    GridClusterDStream containedCluster = null;
	    Pair<Integer, Integer> cellIndex = new Pair<Integer, Integer>(cell.getX(), cell.getY());
	    
	    for (GridClusterDStream cluster : gridClusters) {
	        if (cluster.contain(cellIndex)) {
	            containedCluster = cluster;
	            break;
	        }
	    }
	    
	    return containedCluster;
	}

    private List<GridCharacteristicVector> getNeighbors(GridCharacteristicVector cell) {
        List<GridCharacteristicVector> neighbors = new ArrayList<GridCharacteristicVector>();
        int x = cell.getX();
        int y = cell.getY();
        
        // TODO array check
        neighbors.add(grid[x - 1][y - 1]);
        neighbors.add(grid[x - 1][y]);
        neighbors.add(grid[x - 1][y + 1]);   
        
        neighbors.add(grid[x][y - 1]);
        neighbors.add(grid[x][y + 1]);   
        
        neighbors.add(grid[x + 1][y - 1]);
        neighbors.add(grid[x + 1][y]);
        neighbors.add(grid[x + 1][y + 1]);   
        
        return neighbors;
    }
	
	/**
	 * Main Routine.
	 * TODO: Make this CLI friendly.
	 * 
	 * @param args - algorithms args.
	 */
	public static void main(String[] args) {
		// Console Logging
	    Logger.getRootLogger().setLevel(Level.INFO);
        ConsoleAppender consoleAppender = new ConsoleAppender(new PatternLayout("[%-4p] [%c] %m%n"));
        Logger.getLogger("br.pucrio.inf.lac").addAppender(consoleAppender);
        Logger.getLogger("com.espertech.esper").addAppender(consoleAppender);

        // Booting
        final DStream dstream = new DStream(args[0]);
        
        // Shell
        Runnable r = new Runnable() {
            @Override
            public void run() {
                Scanner scan = new Scanner(System.in);
                
                while (scan.hasNext()) {
                    String command = scan.nextLine();
                    if (command.contains("log")) {
                        try {
                            dstream.log();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }   
                    }
                }
                
            }
        };
        new Thread(r).start();
		dstream.cluster();
	}

	protected void log() throws IOException {
        // TODO Auto-generated method stub
	    long endTime  = 69000000;
	    long currentTime = 60900000;
	    
	    List<DStreamOutput> outList = new ArrayList<DStreamOutput>();
        PrintWriter pw = new PrintWriter(new BufferedWriter(new FileWriter("dstream" + eps + "x" + minPts + ".csv")));


        System.out.println("output size: " + output.size());
	    while (currentTime < endTime) {
	        if (output.containsKey(currentTime)) {
	            outList = output.get(currentTime);
	        }
            printCurrentClusters(currentTime, pw, outList);
	        currentTime += 1000;
	        pw.flush();
	    }
	    
	    pw.flush();
	    pw.close();
    }
	
    private void printCurrentClusters(long currentSecond, PrintWriter pw, List<DStreamOutput> currentClusters) {
        if (currentClusters.isEmpty()) {
            pw.println(currentSecond + ",NONE");
            System.out.println(currentSecond + ",NONE");
        } else {
            for (DStreamOutput cluster : currentClusters) {
                if (cluster.getSize() != 0) {
                    pw.println(currentSecond + ",DSTREAM," + cluster.getSize() + "," + cluster.getContent());
                    System.out.println(currentSecond + ",DSTREAM," + cluster.getSize() + "," + cluster.getContent());
                }
            }
        }
    }
    
    private class DStreamOutput {
        private int size;
        private String content;
        
        public DStreamOutput() {}
        
        public DStreamOutput(int size, String content) {
            this.size = size;
            this.content = content;
        }
        
        public int getSize() {
            return size;
        }
        
        public String getContent() {
            return content;
        }
    }
}
